package com.nami.gate.router.middleware.mq.consumer.listener;

import com.nami.gate.router.middleware.mq.MqSender;
import com.nami.im.network.model.enums.MqTopicConst;
import com.nami.transport.proto.codec.MqTransMsg;
import com.nami.transport.proto.codec.UpWsMsg;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public class PublishListenerImpl extends AbstractConsumer {

    private static final Logger log = LoggerFactory.getLogger(PublishListenerImpl.class);

    private MqSender sender;

    public PublishListenerImpl(MqSender sender) {
        this.sender = sender;
    }

    @Override
    public String getTopicName() {
        return MqTopicConst.PUBLISH;
    }

    @Override
    public void onMessage(byte[] body) {
        try {
            long start = System.currentTimeMillis();

            MqTransMsg.Gate2RouterPublishMsg routerPublishMsg = MqTransMsg.Gate2RouterPublishMsg.parseFrom(body);
            UpWsMsg.PublishMsg upMsg = UpWsMsg.PublishMsg.parseFrom(routerPublishMsg.getData());
            log.info("router publish msg topic={}", upMsg.getTopic());

            MqTransMsg.Router2LogicPublishMsg trans = MqTransMsg.Router2LogicPublishMsg.newBuilder()
                .setSender(routerPublishMsg.getSender())
                .setQos(routerPublishMsg.getQos())
                .setTimestamp(routerPublishMsg.getTimestamp())
                .setTargetId(upMsg.getTargetId())
                .setData(upMsg.getData())
                .build();
            sender.send(upMsg.getTopic(), trans.toByteArray());
            log.info("router 转发消耗:{}ms", System.currentTimeMillis() - start);
        } catch (Exception ex) {
            ex.printStackTrace();
            log.error("router publish msg error", ex);
        }

    }


}
